package eventBusV2

import (
	"errors"
	"sync"
)

const (
	// 固定 worker 数量，根据实际场景可调整
	numWorkers = 4
	// 任务队列容量，避免占用过多内存
	taskQueueSize = 512
)

// Bus 高性能事件总线，保持按值返回，保证向下兼容
type Bus struct {
	subNode     sync.Map       // topic -> *node 映射
	taskQueue   chan asyncTask // 异步任务队列，用于调度消息发送
	shutdownSig chan struct{}  // 关闭信号，通知 worker 退出
	closed      bool           // 标记 Bus 是否已关闭
	mu          sync.Mutex     // 保护 closed 标志
}

// NewBus 初始化一个新的 Bus
func NewBus() Bus {
	bus := Bus{
		taskQueue:   make(chan asyncTask, taskQueueSize),
		shutdownSig: make(chan struct{}),
	}
	// 启动固定数量的 worker（长生命周期 goroutine）
	for i := 0; i < numWorkers; i++ {
		go bus.worker()
	}
	return bus
}

// asyncTask 封装了向某个订阅者发送消息的任务
type asyncTask struct {
	sub Sub
	msg interface{}
}

// worker 从 taskQueue 中取任务处理消息，接收到关闭信号后退出
func (b *Bus) worker() {
	for {
		select {
		case task, ok := <-b.taskQueue:
			if !ok {
				// taskQueue 已关闭
				return
			}
			task.sub.receive(task.msg)
		case <-b.shutdownSig:
			return
		}
	}
}

// Subscribe 订阅指定 topic，将订阅者 sub 添加到对应的节点中
func (b *Bus) Subscribe(topic string, sub Sub) {
	n, _ := b.subNode.LoadOrStore(topic, NewNode())
	node := n.(*node)
	node.rw.Lock()
	defer node.rw.Unlock()
	node.subs = append(node.subs, sub)
}

// UnSubscribe 取消指定 topic 的订阅，若该 topic 无订阅者则删除该节点
func (b *Bus) UnSubscribe(topic string, sub Sub) {
	n, ok := b.subNode.Load(topic)
	if !ok {
		return
	}
	node := n.(*node)
	node.removeSub(sub)
	if node.SubsLen() == 0 {
		b.subNode.Delete(topic)
	}
}

// Publish 发布消息 msg 到指定 topic 下的所有订阅者，消息采用异步分发方式处理
func (b *Bus) Publish(topic string, msg interface{}) error {
	// 若 topic 不存在则返回错误
	n, ok := b.subNode.Load(topic)
	if !ok {
		return errors.New("topic not exist")
	}
	node := n.(*node)
	// 为减少锁持有时间，先复制一份订阅者切片
	node.rw.RLock()
	subs := append([]Sub(nil), node.subs...)
	node.rw.RUnlock()

	// 如果没有订阅者，直接返回
	if len(subs) == 0 {
		return nil
	}

	// 对每个订阅者，构造任务并送入任务队列（阻塞发送确保消息送出）
	for _, sub := range subs {
		task := asyncTask{sub: sub, msg: msg}
		b.taskQueue <- task
	}
	return nil
}

// PubFunc 返回一个便于闭包调用的发布函数
func (b *Bus) PubFunc(topic string) func(msg interface{}) {
	return func(msg interface{}) {
		b.Publish(topic, msg)
	}
}

// SubsLen 返回指定 topic 的订阅者数量
func (b *Bus) SubsLen(topic string) (int, error) {
	n, ok := b.subNode.Load(topic)
	if !ok {
		return 0, errors.New("topic not exist")
	}
	node := n.(*node)
	node.rw.RLock()
	defer node.rw.RUnlock()
	return node.SubsLen(), nil
}

// Close 关闭 Bus，回收所有资源：通知 worker 退出、关闭任务队列以及所有订阅者通道
func (b *Bus) Close() {
	b.mu.Lock()
	if b.closed {
		b.mu.Unlock()
		return
	}
	b.closed = true
	b.mu.Unlock()

	// 先关闭 shutdownSig 通道，通知所有 worker 退出
	close(b.shutdownSig)
	// 关闭任务队列，确保 worker 在读取时退出
	close(b.taskQueue)
	// 遍历所有 topic，关闭对应的订阅者接收通道
	b.subNode.Range(func(key, value interface{}) bool {
		node := value.(*node)
		node.rw.Lock()
		for _, sub := range node.subs {
			// 关闭订阅者的 out 通道，通知订阅者不再有消息
			close(sub.out)
		}
		node.rw.Unlock()
		return true
	})
}

// node 表示一个主题节点，存储所有订阅该 topic 的订阅者
type node struct {
	subs []Sub
	rw   sync.RWMutex
}

// NewNode 返回一个新的 node 实例
func NewNode() *node {
	return &node{
		subs: []Sub{},
	}
}

// SubsLen 返回当前节点中的订阅者数量
func (n *node) SubsLen() int {
	return len(n.subs)
}

// removeSub 从当前节点中移除订阅者 s
func (n *node) removeSub(s Sub) {
	n.rw.Lock()
	defer n.rw.Unlock()
	idx := n.findSubIdx(s)
	if idx < 0 {
		return
	}
	n.subs = append(n.subs[:idx], n.subs[idx+1:]...)
}

// findSubIdx 返回订阅者 s 在切片中的索引，未找到返回 -1
func (n *node) findSubIdx(s Sub) int {
	for idx, sub := range n.subs {
		if sub == s {
			return idx
		}
	}
	return -1
}

// Sub 表示订阅者，包含用于接收消息的通道
type Sub struct {
	out chan interface{}
}

// NewSub 返回一个新的订阅者，内部通道默认缓冲大小为 1024
func NewSub() Sub {
	return Sub{
		out: make(chan interface{}, 512),
	}
}

// receive 将消息 msg 送入订阅者通道，采用阻塞写入以避免额外 goroutine
func (s *Sub) receive(msg interface{}) {
	s.out <- msg
}

// Out 返回订阅者的接收通道
func (s *Sub) Out() chan interface{} {
	return s.out
}
